#include <iostream>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <atomic>
#include <vector>
#include <memory>
#include <functional>
#include <chrono>
#include <sstream>
#include <iomanip>

// 前向声明
class BufferConnection;
typedef std::shared_ptr<BufferConnection> BufferConnectionPtr;

// 回调函数类型定义
typedef std::function<void(const BufferConnectionPtr&, size_t)> HighWaterMarkCallback;
typedef std::function<void(const BufferConnectionPtr&)> WriteCompleteCallback;

// Buffer类实现
class Buffer {
public:
    Buffer(size_t initialSize = 1024) 
        : capacity_(initialSize) {
    }

    size_t readableBytes() const { 
        return queue_.size(); 
    }

    bool empty() const {
        return queue_.empty();
    }

    void put(const int& item) {
        queue_.push(item);
    }

    int take() {
        int item = queue_.front();
        queue_.pop();
        return item;
    }

    size_t capacity() const {
        return capacity_;
    }

private:
    std::queue<int> queue_;
    const size_t capacity_;
};

// 连接类实现
class BufferConnection : public std::enable_shared_from_this<BufferConnection> {
public:
    BufferConnection(size_t highWaterMark = 1000, 
                    size_t lowWaterMark = 100)
        : highWaterMark_(highWaterMark)
        , lowWaterMark_(lowWaterMark)
        , isReading_(true)
        , buffer_(new Buffer())
        , totalProcessed_(0)
        , paused_(false) {
    }

    bool put(const int& item) {
        std::unique_lock<std::mutex> lock(mutex_);
        
        while (paused_) {
            auto status = notFull_.wait_for(lock, std::chrono::milliseconds(100));
            if (!isReading_ || status == std::cv_status::timeout) {
                return false;
            }
        }

        size_t oldLen = buffer_->readableBytes();
        
        if (oldLen >= highWaterMark_) {
            paused_ = true;
            if (highWaterMarkCallback_) {
                highWaterMarkCallback_(shared_from_this(), oldLen);
            }
            return false;
        }

        buffer_->put(item);
        totalProcessed_++;

        notEmpty_.notify_one();
        return true;
    }

    bool take(int& item) {
        std::unique_lock<std::mutex> lock(mutex_);
        
        while (isReading_ && buffer_->empty()) {
            if (notEmpty_.wait_for(lock, std::chrono::milliseconds(100)) 
                == std::cv_status::timeout) {
                return false;
            }
        }

        if (!isReading_ && buffer_->empty()) {
            return false;
        }

        item = buffer_->take();

        if (paused_ && buffer_->readableBytes() <= lowWaterMark_) {
            paused_ = false;
            if (writeCompleteCallback_) {
                writeCompleteCallback_(shared_from_this());
            }
            notFull_.notify_all();
        }

        return true;
    }

    void setHighWaterMarkCallback(const HighWaterMarkCallback& cb, 
                                size_t highWaterMark) {
        std::lock_guard<std::mutex> lock(mutex_);
        highWaterMarkCallback_ = cb;
        highWaterMark_ = highWaterMark;
    }

    void setWriteCompleteCallback(const WriteCompleteCallback& cb) {
        std::lock_guard<std::mutex> lock(mutex_);
        writeCompleteCallback_ = cb;
    }

    void stopRead() {
        std::lock_guard<std::mutex> lock(mutex_);
        isReading_ = false;
        notEmpty_.notify_all();
        notFull_.notify_all();
    }

    void startRead() {
        std::lock_guard<std::mutex> lock(mutex_);
        isReading_ = true;
    }

    bool isReading() const { 
        return isReading_; 
    }

    struct Stats {
        size_t currentSize;
        size_t highWaterMark;
        size_t lowWaterMark;
        bool isReading;
        bool isPaused;
        size_t totalProcessed;

        std::string toString() const {
            std::stringstream ss;
            ss << "Size: " << std::setw(6) << currentSize
               << " HWM: " << std::setw(6) << highWaterMark
               << " LWM: " << std::setw(6) << lowWaterMark
               << " Reading: " << (isReading ? "Yes" : "No")
               << " Paused: " << (isPaused ? "Yes" : "No")
               << " Total: " << std::setw(8) << totalProcessed;
            return ss.str();
        }
    };

    Stats getStats() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return {
            buffer_->readableBytes(),
            highWaterMark_,
            lowWaterMark_,
            isReading_,
            paused_,
            totalProcessed_
        };
    }

private:
    size_t highWaterMark_;
    size_t lowWaterMark_;
    std::atomic<bool> isReading_;
    std::unique_ptr<Buffer> buffer_;
    size_t totalProcessed_;
    bool paused_;

    mutable std::mutex mutex_;
    std::condition_variable notEmpty_;
    std::condition_variable notFull_;

    HighWaterMarkCallback highWaterMarkCallback_;
    WriteCompleteCallback writeCompleteCallback_;
};

class ProcessManager {
public:
    ProcessManager(size_t producerCount = 3, 
                  size_t consumerCount = 2,
                  size_t highWaterMark = 1000,
                  size_t lowWaterMark = 100)
        : conn_(std::make_shared<BufferConnection>(highWaterMark, lowWaterMark))
        , producerCount_(producerCount)
        , consumerCount_(consumerCount)
        , running_(true) {
        
        conn_->setHighWaterMarkCallback(
            [this](const BufferConnectionPtr& conn, size_t len) {
                std::cout << "High water mark " << len 
                         << " items, pausing producers\n";
            },
            highWaterMark);

        conn_->setWriteCompleteCallback(
            [this](const BufferConnectionPtr& conn) {
                std::cout << "Write complete, resuming producers\n";
            });
    }

    void start() {
        running_ = true;
        startProducers();
        startConsumers();
        startMonitor();
    }

    void stop() {
        running_ = false;
        conn_->stopRead();
        
        for (auto& t : producers_) {
            t.join();
        }
        for (auto& t : consumers_) {
            t.join();
        }
        monitor_.join();

        producers_.clear();
        consumers_.clear();
    }

    void adjustIntervals(int producerMs, int consumerMs) {
        producerInterval_.store(producerMs);
        consumerInterval_.store(consumerMs);
        std::cout << "Adjusted intervals - Producer: " << producerMs 
                  << "ms, Consumer: " << consumerMs << "ms\n";
    }

private:
    void producer(int id) {
        int item = 0;
        int backoff = 1;
        
        while (running_) {
            if (conn_->put(item)) {
                item++;
                backoff = 1;
                std::this_thread::sleep_for(
                    std::chrono::milliseconds(producerInterval_));
            } else {
                std::this_thread::sleep_for(
                    std::chrono::milliseconds(backoff));
                backoff = std::min(backoff * 2, 1000);
            }
        }
    }

    void consumer(int id) {
        int item;
        while (running_) {
            if (conn_->take(item)) {
                std::this_thread::sleep_for(
                    std::chrono::milliseconds(consumerInterval_));
            }
        }
    }

    void monitor() {
        while (running_) {
            auto stats = conn_->getStats();
            std::cout << stats.toString() << std::endl;
            std::this_thread::sleep_for(std::chrono::seconds(1));
        }
    }

    void startProducers() {
        for (size_t i = 0; i < producerCount_; ++i) {
            producers_.emplace_back(&ProcessManager::producer, this, i);
        }
    }

    void startConsumers() {
        for (size_t i = 0; i < consumerCount_; ++i) {
            consumers_.emplace_back(&ProcessManager::consumer, this, i);
        }
    }

    void startMonitor() {
        monitor_ = std::thread(&ProcessManager::monitor, this);
    }

    BufferConnectionPtr conn_;
    const size_t producerCount_;
    const size_t consumerCount_;
    std::atomic<bool> running_;
    std::atomic<int> producerInterval_{10};
    std::atomic<int> consumerInterval_{20};

    std::vector<std::thread> producers_;
    std::vector<std::thread> consumers_;
    std::thread monitor_;
};

int main() {
    try {
        ProcessManager pm(3, 2, 1000, 100);
        pm.start();

        // 模拟不同负载场景
        std::cout << "\nPhase 1: Normal load\n";
        std::this_thread::sleep_for(std::chrono::seconds(5));

        std::cout << "\nPhase 2: High load (faster producers)\n";
        pm.adjustIntervals(5, 20);
        std::this_thread::sleep_for(std::chrono::seconds(5));

        std::cout << "\nPhase 3: Recovery (faster consumers)\n";
        pm.adjustIntervals(20, 5);
        std::this_thread::sleep_for(std::chrono::seconds(5));

        std::cout << "\nPhase 4: Back to normal\n";
        pm.adjustIntervals(10, 20);
        std::this_thread::sleep_for(std::chrono::seconds(5));

        pm.stop();
        std::cout << "Process manager stopped normally\n";
    }
    catch (const std::exception& e) {
        std::cerr << "Error: " << e.what() << std::endl;
        return 1;
    }

    return 0;
}